package graph;

import java.io.File;
import java.util.List;
import java.util.Vector;

import mapreduce.DynamicNumberFormat;
import mapreduce.FlipMapper;
import mapreduce.IntegerFormat;
import mapreduce.JobConf;
import mapreduce.Mapper;
import mapreduce.MultiTask;
import mapreduce.Record;
import mapreduce.RecordFeeder;
import mapreduce.RecordFormatWriter;
import mapreduce.RecordReader;
import mapreduce.RecordWriter;
import mapreduce.Reducer;
import mapreduce.Task;
import mapreduce.Record.RecordPosition;
import mapreduce.Record.RecordPositionType;

/**
 * 
 * 
 * @author foxnono
 * 
 */
public class BuildMapTask extends MultiTask {

	static public File getRelationFile(String relationName,
			RelationDirection direction) {
		return new File(relationName + "." + direction + ".rel");
	}

	static public BuildMapTask getTask(int taskId, File dataFile,
			String relationName) {
		BuildMapTask task = new BuildMapTask(null, taskId);
		task.addTask(IndividualBuildMapTask.getTask(taskId * 1000 + 1, dataFile, relationName,
				RelationDirection.DIRECT));
		task.addTask(IndividualBuildMapTask.getTask(taskId * 1000 + 2, dataFile, relationName,
				RelationDirection.INVERSE));
		return task;
	}

	protected BuildMapTask(JobConf conf, int id) {
		super(conf, id);
	}

	static class IndividualBuildMapTask extends Task {
		static IndividualBuildMapTask getTask(int taskId, File dataFile,
				String relationName, RelationDirection direction) {
			RecordFeeder feeder = new RecordReader(dataFile);
			RecordWriter writer = new RecordFormatWriter();
			writer.open(getRelationFile(relationName, direction));
			Mapper mapper = new BuildMapTaskMapper(direction);
			Reducer reducer = new BuildMapTaskReducer();
			JobConf conf = new JobConf(feeder, writer, mapper, reducer);
			IndividualBuildMapTask task = new IndividualBuildMapTask(conf,
					taskId);
			return task;
		}

		private IndividualBuildMapTask(JobConf conf, int taskId) {
			super(conf, taskId);
		}
	}

	static class BuildMapTaskMapper implements Mapper {
		private RelationDirection m_direction;

		public BuildMapTaskMapper(RelationDirection direction) {
			m_direction = direction;
		}

		@Override
		public void map(Record record, Vector<Record> outRecords) {
			Object[] keyData = (Object[]) record.getKey().getData();
			Record.RecordData key = null;
			Record.RecordData value = null;
			if (m_direction == RelationDirection.DIRECT) {
				key = new Record.RecordData(keyData[0], new IntegerFormat());
				value = new Record.RecordData(keyData[1], new IntegerFormat());
			} else {
				key = new Record.RecordData(keyData[1], new IntegerFormat());
				value = new Record.RecordData(keyData[0], new IntegerFormat());
			}
			outRecords.add(Record.getRecord(key, value));
		}

		@Override
		public void release(Vector<Record> outRecords) {
		}
	}
	
	static class BuildMapTaskReducer implements Reducer{

		@Override
		public void reduce(Vector<Record> inRecords, Vector<Record> outRecords) {
			Record.RecordData key = inRecords.get(0).getKey();
			List<Integer> valueList = new Vector<Integer>();
			System.out.printf("%d:", key.getData());
			for(Record record: inRecords){
				valueList.add((Integer) record.getValue().getData());
				valueList.add(new Integer(1));
				System.out.printf("%d ", record.getValue().getData());
			}
			System.out.println();
			
			Record.RecordData value = new Record.RecordData(valueList.toArray(new Integer[0]), new DynamicNumberFormat(new IntegerFormat()));
			outRecords.add(Record.getRecord(key, value));
		}

		@Override
		public void release(Vector<Record> outRecords) {
		}
		
	}

}
